/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.news.server.util;

import cn.ac.iie.di.ban.news.server.exception.NoSuchDataTypeException;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import com.alibaba.fastjson.JSONObject;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author Black-Laptop
 */
public class DataSender {

    private static final Logger logger = LoggerFactory.getLogger(DataSender.class);

    private static final String TYPE_FIELD_NAME = "msg_type";
    private static final String DATA_KEY_FIELD_NAME = "key";
    private static final String DATA_FIELD_NAME = "record_list";

    private static final HashSet<String> GLOBAL_FIELDS;

    static {
        GLOBAL_FIELDS = new HashSet<>();
        GLOBAL_FIELDS.add(TYPE_FIELD_NAME);
        GLOBAL_FIELDS.add(DATA_KEY_FIELD_NAME);
        GLOBAL_FIELDS.add(DATA_FIELD_NAME);
        GLOBAL_FIELDS.add("total_pushed_num");
    }

    private final HashMap<Integer, SenderPool> senderpools;
    private final HashMap<Integer, Map<String, REFieldType>> typeMaps;

    public DataSender() {
        this.senderpools = new HashMap<>();
        this.typeMaps = new HashMap<>();
    }

    public boolean regSender(int type, String namesrv, String topic, Map<String, REFieldType> typeMap, int maxParrallel) throws REConnectionException {
        //exist
        if (senderpools.containsKey(type)) {
            return false;
        }
        //not exist
        senderpools.put(type, new SenderPool(namesrv, topic, typeMap, maxParrallel));
        typeMaps.put(type, typeMap);
        return true;
    }

    public boolean regSender(int type, String namesrv, String topic, Map<String, REFieldType> typeMap) throws REConnectionException {
        //exist
        if (senderpools.containsKey(type)) {
            return false;
        }
        //not exist
        senderpools.put(type, new SenderPool(namesrv, topic, typeMap));
        typeMaps.put(type, typeMap);
        return true;
    }

    public void sendMessage(JSONObject obj) throws RESessionException, REConnectionException, NoSuchDataTypeException {
        logger.debug("start send msg to rmq, data: " + obj.toJSONString());
        //confirm type
        Integer type;
        try {
            type = obj.getInteger(TYPE_FIELD_NAME);
        } catch (Exception e) {
            return;
        }
        if (type == null) {
            return;
        }
        if (!senderpools.containsKey(type)) {
//			throw new NoSuchDataTypeException("no data type \"" + type + "\".");
            logger.error("No rmq sender can be used for data type " + type + " ,maybe data type not registed!");
            return;
        }
        //borrow sender
        Map<String, REFieldType> typeMap = typeMaps.get(type);
        RESendSession sender = senderpools.get(type).Borrow();
        //user desc
        HashMap<String, String> userDesc = new HashMap<>();
        for (String col : GLOBAL_FIELDS) {
            if (typeMap.containsKey(col)) {
                userDesc.put(col, obj.getString(col));
            }
        }
        if (userDesc.isEmpty()) {
        } else {
            sender.setUserDesc(userDesc);
        }
        //data
        try {
            for (Object rec : obj.getJSONArray(DATA_FIELD_NAME)) {
                for (String col : GLOBAL_FIELDS) {
                    if (typeMap.containsKey(col)) {
                        String val = obj.getString(col);
                        switch (typeMap.get(col)) {
                            case String:
                                sender.setString(col, val);
                                break;
                            case Int:
                                sender.setInt(col, Integer.parseInt(val));
                                break;
                            case Long:
                                sender.setLong(col, Long.parseLong(val));
                                break;
                        }
                    }
                }
                for (Map.Entry<String, Object> o : ((JSONObject) rec).entrySet()) {
                    String col = o.getKey();
                    if (typeMap.containsKey(col)) {
                        String val = o.getValue().toString();
                        switch (typeMap.get(col)) {
                            case String:
                                sender.setString(col, val);
                                break;
                            case Int:
                                sender.setInt(col, Integer.parseInt(val));
                                break;
                            case Long:
                                sender.setLong(col, Long.parseLong(val));
                                break;
                        }
                    }
                }
                try {
                    sender.add();
                } catch (Exception e) {
                    while (true) {
                        try {
                            sender.flush();
                            break;
                        } catch (Exception ex) {
                        }
                    }
                    sender.add();
                }
            }
            sender.flush();
            logger.debug("send to rmq succeeded.");
        } finally {
            senderpools.get(type).Return(sender);
        }
    }

}
